package com.gss.plugin.kafka.config;

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;

import javax.annotation.Resource;

/**
 * @author: DYang
 * @since: 2023/8/24
 * @description:
 */
@Slf4j
@Component
public class ProducerHandler {

    @Resource
    private KafkaTemplate<String,String> kafkaTemplate;


    /**
     * 发送消息到kafka
     * @param topic
     * @param data
     */
    public void sendMessage(String topic,Object data) {
        try {
            Assert.notNull(topic,"topic 不能为null");
            Assert.notNull(data, "数据不能为null");
            kafkaTemplate.send(topic, JSONObject.toJSONString(data));
        } catch (Exception e) {
            e.printStackTrace();
            log.error("【Kafka生产者】 发送数据到kafka ",e);
        }
    }

    /**
     * 指定分组发送 topic
     * @param topic
     * @param partition
     * @param key
     * @param data
     */
    public void sendMessageByPartition(String topic,int partition,String key,Object data){
        try {
            Assert.notNull(topic,"topic 不能为null");
            Assert.notNull(data, "数据不能为null");
            Assert.notNull(key, "key不能为null");
            kafkaTemplate.send(topic,partition,key,JSONObject.toJSONString(data));
        } catch (Exception e) {
            e.printStackTrace();
            log.error("【Kafka生产者】 发送数据到指定分区异常 ",e);
        }
    }

    /**
     * 指定分组发送 topic
     * @param topic
     * @param key
     * @param data
     */
    public void sendMessageByKey(String topic,String key,Object data){
        try {
            Assert.notNull(topic,"topic 不能为null");
            Assert.notNull(data, "数据不能为null");
            Assert.notNull(key, "key不能为null");
            kafkaTemplate.send(topic,key,JSONObject.toJSONString(data));
        } catch (Exception e) {
            e.printStackTrace();
            log.error("【Kafka生产者】 指定key发送数据异常 ",e);
        }
    }

}
